use axum::{extract::State, response::sse::{Event, KeepAlive, Sse}};
use serde_json::{json, Value};
use state::AppState;
use tokio::time::interval;
use std::{convert::Infallible, sync::Arc, time::Duration};
use tokio_stream::{wrappers::{IntervalStream, ReceiverStream}, StreamExt as _} ;
use futures_util::stream::{self, Stream};

pub mod state;

pub async fn sse_repeat_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let stream = stream::repeat_with(move || Event::default().data("12313"))
    .map(Ok)
    .throttle(Duration::from_secs(1));
    Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(1)).text("keep-alive-text"))
    // Sse::new(stream).keep_alive(
    //     axum::response::sse::KeepAlive::new()
    //         .interval(Duration::from_secs(1))
    //         .text("keep-alive-text"),
    // )
}

pub async fn sse_interval_handler() -> Sse<impl stream::Stream<Item = Result<Event, axum::Error>>> {
    let interval = interval(Duration::from_secs(1));
    // 使用 tokio 的流工具，每秒发送一次数据
    let stream = IntervalStream::new(interval).map(|_| {
        // Ok(Event::default().event("Msg").data("666").comment("消息事件").id("100").retry(Duration::from_secs(10)))
        Ok(Event::default().json_data(json!({"a":1})).unwrap())
    });
    // 使用 axum 自带的 sse 功能
    Sse::new(stream).keep_alive(KeepAlive::default())
}
// str
pub async fn sse_handler(State(state): State<Arc<AppState>>) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let rx = state.tx.add_str_client().await;
    
    let stream = ReceiverStream::<String>::new(rx)
        .map(|f| Ok(Event::default().data(f)));
    
   Sse::new(stream).keep_alive(KeepAlive::default())
}
//
pub async fn sse_json_handler(State(state): State<Arc<AppState>>) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let rx = state.tx.add_client().await;

    let stream = ReceiverStream::<Value>::new(rx)
        .map(|f| Ok(Event::default().json_data(f).unwrap_or_default()));

    Sse::new(stream)
}